package com.sharif.cn.distributed.zookeeper;

import com.sun.xml.internal.messaging.saaj.util.ByteInputStream;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.*;
import java.util.concurrent.CountDownLatch;

public class ZookeeperPublish implements Watcher {
    private static CountDownLatch latch = new CountDownLatch(1);
    private static Stat stat = new Stat();
    private static ZooKeeper zk = null;
    private final static Integer SESSION_TIMEOUT = Integer.MAX_VALUE;
    private String path;

    public ZookeeperPublish() throws IOException, InterruptedException {
        path = "/publish";
        zk = new ZooKeeper("127.0.0.1:2181", SESSION_TIMEOUT, this);
        latch.await();
    }

    public void publish(Object obj) throws IOException, InterruptedException, KeeperException {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(obj);
        oos.flush();
        byte[] bytes = bos.toByteArray ();
        oos.close();
        bos.close();
        zk.setData(path, bytes, -1);
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        if (Event.KeeperState.SyncConnected == watchedEvent.getState()) {
            System.out.println("receive watched event:" + watchedEvent);
            System.out.println(watchedEvent.getState());
            latch.countDown();
        }
    }
}